package com.alison.sparkstream.action

import com.alibaba.druid.pool.DruidDataSourceFactory
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}

import java.sql.{Connection, PreparedStatement, ResultSet}
import java.text.SimpleDateFormat
import java.util.{Date, Properties}
import javax.sql.DataSource


object E2_Black_list {

  """
    |广告黑名单
    |  需求说明
    |  
    |  （1）我们需要实现实时的动态黑名单机制，即将每天对某个广告点击超过 100 次的用户拉黑（黑名单保存到 MySQL 中）
    |  （2）该需求实现步骤如下：
    |  读取 Kafka 数据之后，对 MySQL 中存储的黑名单数据做校验。
    |  校验通过则对给用户点击广告次数累加 1 并存入 MySQL。
    |  存入 MySQL 之后再对数据做校验，如果单日超过 100 次则将该用户加入黑名单。
    |
    |1. prepare data
    |
    |use spark;
    |CREATE TABLE black_list (userid CHAR(1) PRIMARY KEY) engine=innodb default charset=utf8;
    |
    |CREATE TABLE user_ad_count (
    |  dt varchar(255),
    |  userid CHAR (1),
    |  adid CHAR (1),
    |  count BIGINT,
    |  PRIMARY KEY (dt, userid, adid)
    |) engine=innodb default charset=utf8;
    |
    |2. kafkaUtil
    |
    |3. JdbcUtil
    |
    |4. BlackListHandler 用于处理广告点击数据的工具类, 添加黑名单和过滤黑名单
    |""".stripMargin

  """
    |它读取 Kafka 中的广告点击日志数据流，对数据进行过滤、统计和黑名单处理
    |""".stripMargin

  def main(args: Array[String]): Unit = {
    //1.创建 SparkConf
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RealTimeApp")

    //2.创建 StreamingContext
    val ssc = new StreamingContext(sparkConf, Seconds(3))

    //3.读取 Kafka 数据 1583288137305 华南 深圳 4 3
    val kafkaDStream: InputDStream[ConsumerRecord[String, String]] =
      MyKafkaUtil.getKafkaStream("spark-topic", ssc)

    //4.将每一行数据转换为样例类对象
    val adsLogDStream: DStream[Ads_log] = kafkaDStream.map(record => {
      //a.取出 value 并按照" "切分
      val arr: Array[String] = record.value().split(" ")
      //b.封装为样例类对象
      Ads_log(arr(0).toLong, arr(1), arr(2), arr(3), arr(4))
    })

    //5.根据 MySQL 中的黑名单过滤当前数据集
    val filterAdsLogDStream: DStream[Ads_log] =
      BlackListHandler.filterByBlackList(adsLogDStream)

    //6.将满足要求的用户写入黑名单
    BlackListHandler.addBlackList(filterAdsLogDStream)

    //7.测试打印
    filterAdsLogDStream.cache()
    filterAdsLogDStream.count().print()

    //8.开启任务
    ssc.start()
    ssc.awaitTermination()
  }

  """
    |表示每次广告点击的时间、地区、城市、用户 ID、广告 ID
    |""".stripMargin

  case class Ads_log(timestamp: Long, area: String, city: String, userid: String, adid: String)

  """
    |用于创建一个读取 Kafka 数据的  MyKafkaUtil 工具类
    |""".stripMargin

  object MyKafkaUtil {
    // kafka 消费者配置
    val kafkaParam = Map(
      "bootstrap.servers" -> "localhost:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      //消费者组
      "group.id" -> "commerce-consumer-group",
      //如果没有初始化偏移量或者当前的偏移量不存在任何服务器上，可以使用这个配置属性
      //可以使用这个配置，latest 自动重置偏移量为最新的偏移量
      "auto.offset.reset" -> "latest",
      //如果是 true，则这个消费者的偏移量会在后台自动提交,但是 kafka 宕机容易丢失数据
      //如果是 false，会需要手动维护 kafka 偏移量
      "enable.auto.commit" -> (true: java.lang.Boolean)
    )

    // 创建 DStream，返回接收到的输入数据
    // LocationStrategies：根据给定的主题和集群地址创建 consumer
    // LocationStrategies.PreferConsistent：持续的在所有 Executor 之间分配分区
    // ConsumerStrategies：选择如何在 Driver 和 Executor 上创建和配置 Kafka Consumer
    // ConsumerStrategies.Subscribe：订阅一系列主题
    def getKafkaStream(topic: String, ssc: StreamingContext):
      InputDStream[ConsumerRecord[String, String]] = {
      val dStream: InputDStream[ConsumerRecord[String, String]] =
        KafkaUtils.createDirectStream[String, String](
          ssc,
          LocationStrategies.PreferConsistent,
          ConsumerStrategies.Subscribe[String, String](Array(topic), kafkaParam))
      dStream
    }
  }

  object JdbcUtil {
    //初始化连接池
    var dataSource: DataSource = init()

    //初始化连接池方法
    def init(): DataSource = {
      val properties = new Properties()
      properties.setProperty("driverClassName", "com.mysql.cj.jdbc.Driver")
      properties.setProperty("url", "jdbc:mysql://localhost:3306/spark")
      properties.setProperty("username", "root")
      properties.setProperty("password", "root")
      DruidDataSourceFactory.createDataSource(properties)
    }

    //获取 MySQL 连接
    def getConnection: Connection = {
      dataSource.getConnection
    }

    //执行 SQL 语句,单条数据插入
    def executeUpdate(connection: Connection, sql: String, params: Array[Any]): Int = {
      var returnNum = 0
      var pstmt: PreparedStatement = null
      try {
        connection.setAutoCommit(false)
        pstmt = connection.prepareStatement(sql)
        if (params != null && params.length > 0) {
          for (i <- params.indices) {
            pstmt.setObject(i + 1, params(i))
          }
        }
        returnNum = pstmt.executeUpdate()
        connection.commit()
        pstmt.close()
      } catch {
        case e: Exception => e.printStackTrace()
      }
      returnNum
    }

    //执行 SQL 语句,批量数据插入
    def executeBatchUpdate(connection: Connection, sql: String, paramsList: Iterable[Array[Any]]): Array[Int] = {
      var returnNum: Array[Int] = null
      var pstmt: PreparedStatement = null
      try {
        connection.setAutoCommit(false)
        pstmt = connection.prepareStatement(sql)
        for (params <- paramsList) {
          if (params != null && params.length > 0) {
            for (i <- params.indices) {
              pstmt.setObject(i + 1, params(i))
            }
            pstmt.addBatch()
          }
        }

        returnNum = pstmt.executeBatch()
        connection.commit()
        pstmt.close()
      } catch {
        case e: Exception => e.printStackTrace()
      }
      returnNum
    }

    //判断一条数据是否存在
    def isExist(connection: Connection, sql: String, params: Array[Any]): Boolean = {
      var flag: Boolean = false
      var pstmt: PreparedStatement = null
      try {
        pstmt = connection.prepareStatement(sql)
        for (i <- params.indices) {
          pstmt.setObject(i + 1, params(i))
        }
        flag = pstmt.executeQuery().next()
        pstmt.close()
      } catch {
        case e: Exception => e.printStackTrace()
      }
      flag
    }

    //获取 MySQL 的一条数据
    def getDataFromMysql(connection: Connection, sql: String, params: Array[Any]):
    Long = {
      var result: Long = 0L
      var pstmt: PreparedStatement = null
      try {
        pstmt = connection.prepareStatement(sql)
        for (i <- params.indices) {
          pstmt.setObject(i + 1, params(i))
        }
        val resultSet: ResultSet = pstmt.executeQuery()
        while (resultSet.next()) {
          result = resultSet.getLong(1)
        }
        resultSet.close()
        pstmt.close()
      } catch {
        case e: Exception => e.printStackTrace()
      }
      result
    }
  }

  """
    |用于处理广告点击数据的工具类 BlackListHandler，主要涉及两个功能：添加黑名单和过滤黑名单。
    |""".stripMargin

  object BlackListHandler {
    // 时间格式化对象
    private val sdf = new SimpleDateFormat("yyyy-MM-dd")

    // 添加黑名单
    def addBlackList(filterAdsLogDSteam: DStream[Ads_log]): Unit = {
      // 1.统计当前批次中单日每个用户点击每个广告的总次数
      // 将数据接转换结构 ads_log=>((date,user,adid),1)
      val dateUserAdToOne: DStream[((String, String, String), Long)] =
        filterAdsLogDSteam.map(adsLog => {
          //a.将时间戳转换为日期字符串
          val date: String = sdf.format(new Date(adsLog.timestamp))
          //b.返回值
          ((date, adsLog.userid, adsLog.adid), 1L)
        })

      // 2.统计单日每个用户点击每个广告的总次数
      // 将数据接转换结构 ((date,user,adid),1)=>((date,user,adid),count)
      val dateUserAdToCount: DStream[((String, String, String), Long)] =
        dateUserAdToOne.reduceByKey(_ + _)
      dateUserAdToCount.foreachRDD(rdd => {
        rdd.foreachPartition(iter => {
          val connection: Connection = JdbcUtil.getConnection
          iter.foreach {
            case ((dt, user, ad), count) =>
            // 将该用户对该广告的点击次数进行累加
            JdbcUtil.executeUpdate(connection,
              """
                |INSERT INTO user_ad_count (dt,userid,adid,count)
                |VALUES (?,?,?,?)
                |ON DUPLICATE KEY
                |UPDATE count=count+?
            """.stripMargin, Array(dt, user, ad, count, count))

            // 获取该用户点击该广告的总次数
            val ct: Long = JdbcUtil.getDataFromMysql(connection,
              "select count from user_ad_count where dt=? and userid=? and adid =?",
              Array(dt, user, ad))

            // 如果点击次数超过等于 30 次，就将该用户加入到黑名单表中
            if (ct >= 30) {
              JdbcUtil.executeUpdate(connection,
                "INSERT INTO black_list (userid) VALUES (?) ON DUPLICATE KEY update userid=?",
                Array(user, user))
            }
          }
          connection.close()
        })
      })
    }

    // 过滤黑名单
    def filterByBlackList(adsLogDStream: DStream[Ads_log]): DStream[Ads_log] = {
      // 它将每条广告点击日志的用户ID与黑名单表进行对比
      adsLogDStream.transform(rdd => {
        // 如果用户ID存在于黑名单中，说明该用户被标记为黑名单用户，该条广告点击日志将被过滤掉，不会继续处理
        rdd.filter(adsLog => {
          val connection: Connection = JdbcUtil.getConnection
          val bool: Boolean = JdbcUtil.isExist(connection,
            "select * from black_list where userid=?", Array(adsLog.userid))
          connection.close()
          !bool
        })
      })
    }
  }
}
